from datetime import timedelta
from time import time
from jms_hi.dm.dm_truck_wabill_flow_count_ht import jms_hi_dm__dm_truck_wabill_flow_count_ht
from utils.operators.rich_sql_sensor import RichSqlSensor

label = f"jms_dws_truck_wabill_flow_count_{int(time())}"
timeout = timedelta(minutes=30).seconds
doris_jms_dws__dws_truck_wabill_flow_count = RichSqlSensor(
    task_id='doris_jms_dws__dws_truck_wabill_flow_count',
    pool='broker_load_pool',
    email=['guoruiling@jtexpress.com'],
    conn_id='doris',
    pre_sql=f"""
    LOAD LABEL jms_dws.{label} (
        DATA INFILE(
                    "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_truck_wabill_flow_count_ht/dt={{{{ execution_date | date_add(-3) | cst_ds }}}}/*",
                    "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_truck_wabill_flow_count_ht/dt={{{{ execution_date | date_add(-2) | cst_ds }}}}/*",
                    "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_truck_wabill_flow_count_ht/dt={{{{ execution_date | date_add(-1) | cst_ds }}}}/*",
                    "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_truck_wabill_flow_count_ht/dt={{{{ execution_date | cst_ds }}}}/*"
                    )
        INTO TABLE dws_truck_wabill_flow_count  
      FORMAT AS 'PARQUET'
     (shipment_no,previous_network_code,current_center_code,next_network_code,previous_network_name,current_center_name,next_network_name,waybill_cnt,same_next_cnt,rk)
     columns from path as (dt)
     SET (dt=dt) 
     )
    WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
    PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM jms_dws WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_dws WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120), )

doris_jms_dws__dws_truck_wabill_flow_count << jms_hi_dm__dm_truck_wabill_flow_count_ht